package discov

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/ameise84/e3dds/internal/log"
	"gitee.com/ameise84/e3pool/go_pool"
	"gitee.com/ameise84/e3utils/safe"
	"gitee.com/ameise84/e3utils/str_conv"
	"gitee.com/ameise84/e3utils/sys"
	"github.com/pkg/errors"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

var (
	dialTTL = 10 * time.Second
)

func newEtcd() driver {
	dr := &etcd{}
	dr.gwg = go_pool.NewGoRunner(dr, "gwg", go_pool.DefaultOptions().SetSimCount(0))
	return dr
}

type etcd struct {
	running atomic.Bool
	s       *concurrency.Session
	cli     *clientv3.Client
	gwg     go_pool.GoRunner
}

func (ts *etcd) E3LogMarshall() string {
	return "etcd"
}

func (ts *etcd) OnPanic(err error) {
	log.GetLogger().Error("OnPanic").Object(ts).Err(err).Println()
}

func (ts *etcd) start(opts *Options) error {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   opts.addr,
		DialTimeout: dialTTL,
	})
	if err != nil {
		return errors.WithMessage(err, ts.E3LogMarshall())
	}
	ts.s, err = concurrency.NewSession(cli, concurrency.WithTTL(opts.keepAlive))
	if err != nil {
		return errors.WithMessage(err, ts.E3LogMarshall())
	}
	ts.s.Orphan()
	ts.cli = ts.s.Client()
	ts.manualKeepAlive(ts.cli, ts.s.Lease(), int64(opts.keepAlive))
	ts.running.Store(true)
	return nil
}

func (ts *etcd) stop() {
	err := ts.s.Close()
	if err != nil {
		log.GetLogger().Error("close etcd session error").Object(ts).Err(err)
	}
	err = ts.cli.Close()
	if err != nil {
		log.GetLogger().Error("close etcd client error").Object(ts).Err(err)
		return
	}
	ts.running.Store(false)
	ts.gwg.Wait()
	ts.s = nil
	ts.cli = nil
}

func (ts *etcd) watch(cb WatchHandler, k string, id int64) (err error) {
	ctx, c := context.WithTimeout(ts.cli.Ctx(), dialTTL)
	var wc clientv3.WatchChan

	go safe.Func(nil, "watch", func() {
		wc = ts.cli.Watch(ts.cli.Ctx(), k, clientv3.WithPrevKV(), clientv3.WithPrefix(), clientv3.WithRev(id))
		c()
	})

	<-ctx.Done()
	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
		return errors.WithMessage(ctx.Err(), "timeout")
	}

	_ = ts.gwg.AsyncRun(
		func(...any) {
		loopFor:
			for {
				select {
				case <-ts.cli.Ctx().Done():
					break loopFor
				case wr := <-wc:
					cb.OnWatch(ts.transformEvents(wr.Events))
				}
			}
		})
	return
}

func (ts *etcd) update(k, v string) (err error) {
	ctx, c := context.WithTimeout(ts.cli.Ctx(), dialTTL)
	defer c()

	_, err = ts.cli.Put(ctx, k, v, clientv3.WithLease(ts.s.Lease()))
	if err != nil {
		return errors.WithMessage(err, "update")
	}
	return nil
}

func (ts *etcd) put(k, v string) (err error) {
	ctx, c := context.WithTimeout(ts.cli.Ctx(), dialTTL)
	defer c()
	_, err = ts.cli.Put(ctx, k, v, clientv3.WithLease(ts.s.Lease()))
	if err != nil {
		return errors.WithMessage(err, "put")
	}
	return
}

func (ts *etcd) get(k string) (int64, []string, error) {
	ctx, c := context.WithTimeout(ts.cli.Ctx(), dialTTL)
	defer c()
	rsp, err := ts.cli.Get(ctx, k, clientv3.WithPrefix())
	if err != nil {
		return 0, nil, errors.WithMessage(err, "get")
	}
	var out []string
	for _, v := range rsp.Kvs {
		out = append(out, string(v.Value))
	}
	return rsp.Header.Revision, out, nil
}

func (ts *etcd) del(k string) (string, error) {
	ctx, c := context.WithTimeout(ts.cli.Ctx(), dialTTL)
	defer c()
	rsp, err := ts.cli.Delete(ctx, k, clientv3.WithPrevKV())
	if err != nil || rsp.PrevKvs == nil {
		return "", errors.WithMessage(err, "delete")
	}
	return string(rsp.PrevKvs[0].Value), nil
}

func (ts *etcd) getLocker(k string) sync.Locker {
	return concurrency.NewLocker(ts.s, k)
}

func (ts *etcd) lockDo(k string, f func() error) (err error) {
	safe.Func(nil, "etcd lock do", func() {
		l := concurrency.NewLocker(ts.s, k)
		l.Lock()
		err = f()
		l.Unlock()
	})
	return err
}

func (ts *etcd) transformEvents(old []*clientv3.Event) []Event {
	out := make([]Event, 0, len(old))
	for _, r := range old {
		out = append(out, ts.transformEven(r))
	}
	return out
}

func (ts *etcd) transformEven(old *clientv3.Event) (evt Event) {
	evt.Kind = EventDel
	if old.IsCreate() {
		evt.Kind = EventCreate
		evt.Key = str_conv.ToString(old.Kv.Key)
		evt.Value = str_conv.ToString(old.Kv.Value)
	} else if old.IsModify() {
		evt.Kind = EventModify
		evt.Key = str_conv.ToString(old.Kv.Key)
		evt.Value = str_conv.ToString(old.Kv.Value)
	} else {
		evt.Key = str_conv.ToString(old.PrevKv.Key)
		evt.Value = str_conv.ToString(old.PrevKv.Value)
	}
	return
}

func (ts *etcd) manualKeepAlive(cli *clientv3.Client, leaseID clientv3.LeaseID, ttl int64) {
	go safe.Func(ts, "etcd keep alive", func() {
		keep := 120 * time.Second
		if ttl <= 120/2*3 {
			keep = time.Duration(ttl) * time.Second * 2 / 3
		}
		sleep := keep / 10

		ticker := time.NewTicker(keep)
		defer ticker.Stop()

	loopKeep:
		for {
			select {
			case <-ticker.C:
				times := 0
				ok := false
				for times < 5 {
					_, err := cli.KeepAliveOnce(cli.Ctx(), leaseID)
					if err == nil {
						ok = true
						break
					}
					log.GetLogger().Warn("keep alive error").Int64("lease id", int64(leaseID)).Err(err).Println()
					times++
					time.Sleep(sleep)
				}
				if !ok {
					if ts.running.Load() {
						log.GetLogger().Error("keep alive failed").Int64("lease id", int64(leaseID)).Println()
					}
					sys.Exit(-100)
				}
			case <-ts.cli.Ctx().Done():
				break loopKeep
			}
		}
	})
}
